package baseFx.common.mq.rabbitmq;

import baseFx.common.mq.MqPublisher;
import baseFx.common.utils.ExceptionUtils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.MessageProperties;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeoutException;

public class RabbitMqPublisher implements MqPublisher {
    private static List<String> exists = new ArrayList<>();
    private Connection connection;
    private Channel channel;
    private RabbitMqContext context;

    public RabbitMqPublisher(RabbitMqContext context) {
        this.context = context;
        try {
            this.connection = context.getConnection();
            this.channel = this.connection.createChannel();
        } catch (IOException e) {
            throw ExceptionUtils.toRuntime(e);
        }
    }

    public void post(String exchange, String routing, String[] msg) {
        post(exchange + "&" + routing, Arrays.asList(msg));
    }

    @Override
    public void post(String queue, List<String> msg) {
        if (msg == null || msg.size() < 1) {
            return;
        }
        if (exists.contains(queue) == false) {
            if (context.existsExchange(queue) == false) {
                context.exchangeDeclare(queue, BuiltinExchangeType.FANOUT);
            }
            exists.add(queue);
        }
        String[] pam = queue.split("&");
        if (pam.length == 0 || pam.length > 2) {
            throw new RuntimeException("queue的正确格式是：<exchange>[&routing]");
        }
        try {
            String r = pam.length == 1 ? "" : pam[1];
            for (String item : msg) {
                channel.basicPublish(pam[0], r, MessageProperties.PERSISTENT_TEXT_PLAIN, item.getBytes(StandardCharsets.UTF_8));
            }
        } catch (IOException e) {
            throw ExceptionUtils.toRuntime(e);
        }
    }

    @Override
    public void close() {
        if (channel != null) {
            try {
                channel.close();
            } catch (IOException | TimeoutException e) {
                e.printStackTrace();
            } finally {
                channel = null;
            }
        }
        if (connection != null) {
            try {
                connection.close();
            } catch (IOException e) {
                e.printStackTrace();
            } finally {
                connection = null;
            }
        }
        context = null;
    }
}
